package com.sduept.component.message.core;

import com.sduept.nwld.dataserver.manager.etl.*;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class BxEquipmentService {


    @Autowired
    private BxAclinesegmentETLManager aclineManager;

    @Autowired
    private BxBreakerETLManager breakerManager;

    @Autowired
    private BxBusbarsectionETLManager busbarManager;

    @Autowired
    private BxCompensatorETLManager comManager;

    @Autowired
    private BxHydroGeneratingUnitETLManager unitManager;

    @Autowired
    private BxPowertransformerETLManager powerManager;

    @Autowired
    private BxSubStationETLManager stationManager;

    private ExecutorService executor = Executors.newFixedThreadPool(10);

    @KafkaListener(topics = {"bx_equipment_update"})
    public void onMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            JSONObject jo = JSONObject.fromObject(kafkaMessage.get());
            String msgContent = jo.getString("content");
            JSONObject msgJson = JSONObject.fromObject(msgContent);
            final String datetime = msgJson.getString("updatetime");
            String tableName = msgJson.getString("tableName");
            log.info("接收到消息：bx_equipment_update, 数据表："+tableName+",数据更新时间："+datetime);
            if ("t_bx_aclinesegment".equals(tableName)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        aclineManager.receiveAndSavePartData(datetime);
                    }
                });
            } else if ("t_bx_breaker".equals(tableName)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        breakerManager.receiveAndSavePartData(datetime);
                    }
                });
            } else if ("t_bx_busbarsection".equals(tableName)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        busbarManager.receiveAndSavePartData(datetime);
                    }
                });
            } else if ("t_bx_compensator".equals(tableName)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        comManager.receiveAndSavePartData(datetime);
                    }
                });
            } else if ("t_bx_hydrogeneratingunit".equals(tableName)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        unitManager.receiveAndSavePartData(datetime);
                    }
                });
            } else if ("t_bx_powertransformer".equals(tableName)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        powerManager.receiveAndSavePartData(datetime);
                    }
                });
            } else if ("t_bx_substation".equals(tableName)) {
                executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        stationManager.receiveAndSavePartData(datetime);
                    }
                });
            }
        }


    }
}
